package com.icoopoo.flux.websocket;

import com.alibaba.fastjson.JSON;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
import reactor.core.publisher.Mono;
import reactor.core.publisher.UnicastProcessor;
import reactor.netty.http.client.HttpClient;
import reactor.netty.tcp.TcpClient;

import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;

//因为是用main启动的，所以都是静态方法，实际项目需要修改
public class WebsocketTest {

    //只是为了阻塞主线程，防止退出，所以，真正使用的时候需要去除相关的这个东西
    static CountDownLatch countDownLatch = new CountDownLatch(1);
    //自定义的发送器
    static UnicastProcessor<String> directProcessor = UnicastProcessor.create();

    public static void main(String[] args) {
        websocket();
        HashMap<String, Object> params = new HashMap<>();
        params.put("checkpoint", "0");
        FunctionRequest functionRequest = new FunctionRequest();
        functionRequest.setId(System.currentTimeMillis() + "");
        functionRequest.setDeviceId("gateway1");
        functionRequest.setFunctionId("FETCH_ACCESS_RECORDS");
        functionRequest.setParameter(params);
        directProcessor.onNext(JSON.toJSONString(functionRequest));
        try {
            //Thread.sleep(500000);
            countDownLatch.await();
            System.out.println("----------");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void websocket() {

        ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient(httpClient());
        client.setHandlePing(true);
        client.setMaxFramePayloadLength(65535 * 10);
        try {
            //连接websocket 服务器
            client.execute(new URI("ws://10.20.111.223/jetlinks/function"), session -> {
                Mono<Void> input = session.receive()
                        .doOnNext(webSocketMessage -> {
                            String result = webSocketMessage.getPayloadAsText(StandardCharsets.UTF_8);
                            JSON.parseObject(result);
                            System.out.println("收到消息：" + result);
                             })
                        .doOnError(throwable -> System.out.println("发生异常：" + throwable))
                        .doOnComplete(() -> System.out.println("结束"))
                        .then();
                Mono<Void> output = directProcessor.map(message -> session.textMessage(message))
                        .flatMap(webSocketMessage -> session.send(Mono.just(webSocketMessage))).then();

                return Mono.zip(input, output).then()
                        .doFinally(
                                signalType -> {
                                    System.out.println("连接结束了，不知道是发生的了异常，还是正常断开，如果要重连，可以在这儿想办法再次发起连接");
                                    countDownLatch.countDown();
                                });
            }).onTerminateDetach()
                    .doOnError(throwable -> System.out.println("发生异常:" + throwable))
                    .subscribe(aVoid -> System.out.println("这个打印只是想告诉使用的人，reactor中所有的发布订阅最终操作为订阅操作（subcribe）"));
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    public static HttpClient httpClient() {

        TcpClient tcpClient = TcpClient.create()
                //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
                .doOnConnected(connection ->
                        connection.addHandlerLast(new ReadTimeoutHandler(6000))
                                .addHandlerLast(new WriteTimeoutHandler(6000)));
        return HttpClient.from(tcpClient);
    }

}



